package com.maimao.kill.config;

import com.google.common.collect.Maps;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.util.Map;

/**
 * @author MaoLin Wang
 * @date 2021/1/18 2:43 下午
 */
@Configuration
public class RabbitConfig {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private Environment environment;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;


    private static final String SPRING_RABBITMQ_LISTENER_SIMPLE_CONCURRENCY = "spring.rabbitmq.listener.simple.concurrency";
    private static final String SPRING_RABBITMQ_LISTENER_SIMPLE_MAX_CONCURRENCY = "spring.rabbitmq.listener.simple.max-concurrency";
    private static final String SPRING_RABBITMQ_LISTENER_SIMPLE_PREFETCH = "spring.rabbitmq.listener.simple.prefetch";

    /**
     * 单一消费者
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setBatchSize(1);
        return factory;
    }

    /**
     * 多个消费者
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, cachingConnectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //消费者手动确认
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setConcurrentConsumers(environment.getProperty(SPRING_RABBITMQ_LISTENER_SIMPLE_CONCURRENCY, int.class));
        factory.setMaxConcurrentConsumers(environment.getProperty(SPRING_RABBITMQ_LISTENER_SIMPLE_MAX_CONCURRENCY, int.class));
        factory.setPrefetchCount(environment.getProperty(SPRING_RABBITMQ_LISTENER_SIMPLE_PREFETCH, int.class));
        return factory;
    }


    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

    /**
     * *************************订单**********************************8
     */

    /**
     * 秒杀订单提交延迟队列（不用来消费，而是绑定了死信队列，超时未支付进入死信队列来进行延迟消费） 发送预生成的订单id，消费者收到后生成订单入库
     */
    @Bean
    public Queue killOrderSubmitDelayQueue() {
        Map<String, Object> argsMap = Maps.newHashMap();
        //当前队列绑定的私信交换机
        //消息过期时间
//        argsMap.put("x-message-ttl",1000 * 60 * 2);

        argsMap.put("x-dead-letter-exchange", RabbitProperties.KILL_ORDER_DEAD_EXCHANGE);
        //当前队列绑定的死信路由key
        argsMap.put("x-dead-letter-routing-key", RabbitProperties.KILL_ORDER_DEAD_ROUTING_KEY);
        return new Queue(RabbitProperties.KILL_ORDER_SUBMIT_DELAY_QUEUE, true, false, false, argsMap);
    }

    /**
     * 订单提交延迟交换机
     */
    @Bean
    public TopicExchange killOrderSubmitDelayExchange() {
        return new TopicExchange(RabbitProperties.KILL_ORDER_SUBMIT_DELAY_EXCHANGE, true, false);
    }

    /**
     * 延迟队列绑定延迟交换机+延迟路由
     */
    @Bean
    public Binding killOrderSubmitDelayBinding() {
        return BindingBuilder
                .bind(killOrderSubmitDelayQueue())
                .to(killOrderSubmitDelayExchange())
                .with(RabbitProperties.KILL_ORDER_SUBMIT_DELAY_ROUTING_KEY);
    }

    /**
     * * ======================== 秒杀成功发送消息，进行预生成订单保存到redis =====================
     */
    @Bean
    public Queue killSuccessQueue() {
        return new Queue(RabbitProperties.KILL_SUCCESS_QUEUE, true, false, false);
    }

    /**
     * 秒杀成功交换机
     */
    @Bean
    public TopicExchange killSuccessExchange() {
        return new TopicExchange(
                RabbitProperties.KILL_SUCCESS_EXCHANGE,
                true,
                false);
    }

    /**
     * 秒杀订单提交死信队列绑定死信交换机+死信路由
     */
    @Bean
    public Binding killSuccessBinding() {
        return BindingBuilder
                .bind(killSuccessQueue())
                .to(killSuccessExchange())
                .with(RabbitProperties.KILL_SUCCESS_ROUTING_KEY);
    }

    /**
     * * ======================== 预生成订单发送消息，进行订单入库 =====================
     */
    @Bean
    public Queue killSubmitOrderQueue() {
        return new Queue(RabbitProperties.KILL_ORDER_SUBMIT_QUEUE, true, false, false);
    }

    /**
     * 秒杀订单提交交换机
     */
    @Bean
    public TopicExchange killSubmitOrderExchange() {
        return new TopicExchange(
                RabbitProperties.KILL_ORDER_SUBMIT_EXCHANGE,
                true,
                false);
    }

    /**
     * 秒杀订单提交死信队列绑定死信交换机+死信路由
     */
    @Bean
    public Binding killSubmitOrderBinding() {
        return BindingBuilder
                .bind(killSubmitOrderQueue())
                .to(killSubmitOrderExchange())
                .with(RabbitProperties.KILL_ORDER_SUBMIT_ROUTING_KEY);
    }


    /**
     * ======================== 订单提交绑定的死信队列 =====================
     **/
    /**
     * 秒杀死信队列 缺陷：大批的订单在某个TTL集中失效时，恰好RabbitMQ挂掉， 使用定时任务扫描未支付订单进行取消
     */
    @Bean
    public Queue killOrderDeadQueue() {
        return new Queue(RabbitProperties.KILL_ORDER_DEAD_QUEUE, true);
    }

    /**
     * 秒杀死信交换机
     */
    @Bean
    public TopicExchange killOrderDeadExchange() {
        return new TopicExchange(
                RabbitProperties.KILL_ORDER_DEAD_EXCHANGE,
                true,
                false);
    }

    /**
     * 秒杀订单提交死信队列绑定死信交换机+死信路由
     */
    @Bean
    public Binding killOrderSubmitDeadBinding() {
        return BindingBuilder
                .bind(killOrderDeadQueue())
                .to(killOrderDeadExchange())
                .with(RabbitProperties.KILL_ORDER_DEAD_ROUTING_KEY);
    }


}
